

package com.hazelcast.query.impl;

import com.hazelcast.config.IndexConfig;
import com.hazelcast.instance.impl.Node;
import com.hazelcast.internal.monitor.impl.PartitionIndexChangeEvent;
import com.hazelcast.internal.monitor.impl.PerIndexStats;
import com.hazelcast.internal.serialization.InternalSerializationService;
import com.hazelcast.query.impl.GlobalIndexPartitionTracker.PartitionStamp;
import com.hazelcast.query.impl.getters.Extractors;

/**
 * Provides implementation of on-heap indexes.
 */
public class IndexImpl extends AbstractIndex {

    private final GlobalIndexPartitionTracker partitionTracker;

    public IndexImpl(Node node, IndexConfig config, InternalSerializationService ss, Extractors extractors, IndexCopyBehavior copyBehavior, PerIndexStats stats, int partitionCount, String mapName) {
        super(node, config, ss, extractors, copyBehavior, stats, mapName, GLOBAL_INDEX_NOOP_PARTITION_ID);

        partitionTracker = new GlobalIndexPartitionTracker(partitionCount);
    }

    @Override
    protected IndexStore createIndexStore(Node node, IndexConfig config, PerIndexStats stats, String mapName, int ignored) {
        switch (config.getType()) {
        case SORTED:
            return new OrderedIndexStore(copyBehavior);
        case HASH:
            return new UnorderedIndexStore(copyBehavior);
        case BITMAP:
            return new BitmapIndexStore(config);
        default:
            throw new IllegalArgumentException("unexpected index type: " + config.getType());
        }
    }

    @Override
    public final boolean hasPartitionIndexed(int partitionId) {
        return partitionTracker.isIndexed(partitionId);
    }

    @Override
    public final boolean allPartitionsIndexed(int ownedPartitionCount) {
        // This check guarantees that all partitions are indexed
        // only if there is no concurrent migrations. Check migration stamp
        // to detect concurrent migrations if needed.
        return ownedPartitionCount < 0 || partitionTracker.indexedCount() == ownedPartitionCount;
    }

    @Override
    public final void beginPartitionUpdate() {
        partitionTracker.beginPartitionUpdate();
        getPerIndexStats().onPartitionChange(PartitionIndexChangeEvent.CHANGE_STARTED);
    }

    @Override
    public final void markPartitionAsIndexed(int partitionId) {
        if (partitionTracker.partitionIndexed(partitionId)) {
            getPerIndexStats().onPartitionChange(PartitionIndexChangeEvent.INDEXED);
        }
        getPerIndexStats().onPartitionChange(PartitionIndexChangeEvent.CHANGE_FINISHED);
    }

    @Override
    public final void markPartitionAsUnindexed(int partitionId) {
        if (partitionTracker.partitionUnindexed(partitionId)) {
            getPerIndexStats().onPartitionChange(PartitionIndexChangeEvent.UNINDEXED);
        }
        getPerIndexStats().onPartitionChange(PartitionIndexChangeEvent.CHANGE_FINISHED);
    }

    @Override
    public final void clear() {
        partitionTracker.clear();
        super.clear();
    }

    @Override
    public final PartitionStamp getPartitionStamp() {
        return partitionTracker.getPartitionStamp();
    }

    @Override
    public final boolean validatePartitionStamp(long stamp) {
        return partitionTracker.validatePartitionStamp(stamp);
    }

    @Override
    public String toString() {
        return "IndexImpl{" + "partitionTracker=" + partitionTracker + "} " + super.toString();
    }
}
